package com.xhs.zk;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by jm on 2017/11/27.
 */
public class ZKSession {

    private final String zkHosts = "101.200.183.100:2181";
    private static ZooKeeper zooKeeper;
    private CountDownLatch countDownLatch = new CountDownLatch(1);

    //建立 zk session的watcher
    private class ZookeeperWatcher implements Watcher {
        @Override
        public void process(WatchedEvent watchedEvent) {
            //如果监听到的状态是syncConnected 就释放锁
            if (Event.KeeperState.SyncConnected == watchedEvent.getState())
                countDownLatch.countDown();
            else
                System.out.println("=====   ZK 连接中; 状态:"+watchedEvent.getState());
        }
    }

    public String getNodeData(String path){
        try {
            return new String(zooKeeper.getData(path,false,new Stat()));
        }catch (Exception e){
            e.printStackTrace();
        }
        return null;
    }


    public void setNodeData(String path,String data){
        try {
            zooKeeper.setData(path,data.getBytes(),-1);
        }catch (Exception e){
            e.printStackTrace();
        }
    }


    public void createNode(String path){
        try {
            zooKeeper.create(path,"".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
        } catch (Exception e) {
            System.out.println("");
            //e.printStackTrace();
        }
    }

    /**
     * 获取分布式锁
     *
     * @param key
     */
    public static void acquireDistributedLock(String key){
        //记录下重试次数
        AtomicInteger retries = new AtomicInteger(1);
        //锁的path
        String lockPath = "/distributed-lock-"+key;

        try {
            //创建临时node
            zooKeeper.create(lockPath,"".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        }catch (Exception e) {
            // 已经存在了，就是已经被别人加锁了，那么就这里就会报错
            //会报这样的错org.apache.zookeeper.KeeperException$NodeExistsException: KeeperErrorCode = NodeExists for /distributed-lock-6
//            e.printStackTrace();
            while (true){
                try {
                    Thread.sleep(1000);
                    System.out.println("重试获取锁:["+lockPath+"],重试:"+retries.get());
                    zooKeeper.create(lockPath,"".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                }catch (Exception e1){
                    retries.incrementAndGet();
                    continue;
                }
                System.out.println("我拿到了锁:["+lockPath+"] 共尝试了 ["+retries.get()+"] 次");
                break;
            }
        }
        System.out.println("我拿到了锁:["+lockPath+"]");
    }

    /**
     * 释放锁
     * @param key
     */
    public static void releaseLock(String key){
        //锁的path
        String lockPath = "/distributed-lock-"+key;
        try {
            zooKeeper.delete(lockPath,-1);
        }  catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("释放掉 ["+lockPath+"] 锁");
    }



    private ZKSession(){
        init();
    }

    private static class SingletonHolder{
        private static ZKSession zkSession = new ZKSession();
    }

    public static ZKSession getInstance(){
        return SingletonHolder.zkSession;
    }

    private void init(){
        if (zooKeeper!=null && zooKeeper.getState().isConnected() ) {
            System.out.println("=====   ZK 已经连接成功 不需要初始化  =====");
            return;
        }

        try {
            zooKeeper = new ZooKeeper(zkHosts,60*1000,new ZookeeperWatcher());
            countDownLatch.await();

        }catch (Exception e){
            e.printStackTrace();
            System.out.println("=====   ZK 连接失败   =====");
        }
        if (zooKeeper!=null &&zooKeeper.getState().isConnected())
            System.out.println("=====   ZK 连接成功   =====");
    }


}
